const express = require('express');
const mqtt = require('mqtt');
const cors = require('cors');
const WebSocket = require('ws');
const bodyParser = require('body-parser');
const mysql = require('mysql2/promise');
const path = require('path');
require('dotenv').config();

const app = express();
const PORT = process.env.PORT || 3002;

// 中间件
app.use(cors());
app.use(bodyParser.json());
app.use(express.static(path.join(__dirname, 'client-vue/dist')));

// 使用内存存储日志
const logs = [];

// MySQL 连接池
let mysqlPool = null;
async function initMySQL() {
  try {
    const {
      MYSQL_HOST = '8.149.243.80',  // 正确的云服务器公网IP
      MYSQL_PORT = '3307',
      MYSQL_USER = 'root',
      MYSQL_PASSWORD = '123456',  // 云服务器数据库密码
      MYSQL_DATABASE = 'mqtt_app'
    } = process.env;

    mysqlPool = mysql.createPool({
      host: MYSQL_HOST,
      port: Number(MYSQL_PORT),
      user: MYSQL_USER,
      password: MYSQL_PASSWORD,
      database: MYSQL_DATABASE,
      waitForConnections: true,
      connectionLimit: 10,
      queueLimit: 0,
      idleTimeout: 300000,
      connectTimeout: 5000
      // 移除 acquireTimeout - MySQL2 不支持此配置项
    });

    // 初始化表
    await mysqlPool.query(`CREATE TABLE IF NOT EXISTS logs (
      id BIGINT PRIMARY KEY AUTO_INCREMENT,
      timestamp VARCHAR(32) NOT NULL,
      type VARCHAR(32) NOT NULL,
      topic VARCHAR(255) DEFAULT NULL,
      message MEDIUMTEXT,
      qos TINYINT DEFAULT 0,
      retained TINYINT DEFAULT 0,
      source VARCHAR(32) DEFAULT NULL,
      client_id VARCHAR(128) NULL,
      created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
      updated_at DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
      INDEX idx_ts (timestamp),
      INDEX idx_type (type),
      INDEX idx_topic (topic),
      INDEX idx_created_at (created_at)
    ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;`);

    // 检查并添加source字段（如果不存在）
    try {
      await mysqlPool.query(`
        ALTER TABLE logs
        ADD COLUMN IF NOT EXISTS source VARCHAR(32) DEFAULT NULL
      `);
    } catch (err) {
      // MySQL可能不支持IF NOT EXISTS，使用另一种方法
      try {
        await mysqlPool.query(`
          SELECT source FROM logs LIMIT 1
        `);
      } catch (e) {
        // 字段不存在，添加它
        await mysqlPool.query(`
          ALTER TABLE logs
          ADD COLUMN source VARCHAR(32) DEFAULT NULL
        `);
        console.log('已添加source字段到logs表');
      }
    }

    console.log('MySQL 已连接，日志表就绪');
  } catch (err) {
    console.warn('MySQL 初始化失败，继续使用内存日志:', err.message);
    mysqlPool = null;
    // 继续执行，确保服务器能够正常启动
  }
}

// MQTT客户端状态
let mqttClient = null;
let isConnecting = false;
let backoffUntilTs = 0;
const subscribedTopics = new Set();
let mqttConfig = {
  host: 'broker.emqx.io',
  port: 1883,
  username: '',
  password: '',
  clientId: process.env.MQTT_CLIENT_ID || 'mqtt-client-app',
  keepalive: 30,
  reconnectPeriod: 5000,
  connectTimeout: 10000,
  clean: false,
  protocolVersion: 4,
  secure: false,
  rejectUnauthorized: false,
  ca: '',
  cert: '',
  key: ''
};

// WebSocket服务器
const WS_PORT = process.env.WS_PORT || 8081;
const wss = new WebSocket.Server({ port: WS_PORT });

// 广播消息到所有连接的客户端
function broadcastToClients(data) {
  wss.clients.forEach(client => {
    if (client.readyState === WebSocket.OPEN) {
      client.send(JSON.stringify(data));
    }
  });
}

// 添加日志
async function addLog(type, topic, message, source) {
  const log = {
    timestamp: new Date().toLocaleString('zh-CN'),
    type,
    topic,
    message,
    qos: 0,
    retained: false,
    source: source ? source.substring(0, 32) : null, // 限制source字段长度为32字符
    client_id: mqttConfig.clientId
  };

  logs.unshift(log);
  if (logs.length > 1000) {
    logs.splice(1000);
  }

  // 保存到数据库
  if (mysqlPool) {
    try {
      await mysqlPool.query(
        'INSERT INTO logs (timestamp, type, topic, message, qos, retained, source, client_id) VALUES (?, ?, ?, ?, ?, ?, ?, ?)',
        [log.timestamp, log.type, log.topic, log.message, log.qos, log.retained, log.source, log.client_id]
      );
    } catch (err) {
      console.warn('保存日志到数据库失败:', err.message);
      // 如果数据库连接失败，尝试重新初始化连接
      if (err.code === 'ECONNRESET' || err.code === 'PROTOCOL_CONNECTION_LOST') {
        console.log('数据库连接丢失，尝试重新初始化...');
        setTimeout(() => {
          initMySQL().catch(console.error);
        }, 5000);
      }
    }
  }
}

// 连接MQTT
async function connectMQTT() {
  if (isConnecting) {
    console.warn('MQTT连接正在进行中，跳过重复连接');
    return;
  }

  if (Date.now() < backoffUntilTs) {
    console.warn('MQTT连接冷却中，跳过连接');
    return;
  }

  isConnecting = true;

  try {
    if (mqttClient) {
      mqttClient.end();
    }

    // 根据secure配置选择协议
    const protocol = mqttConfig.secure ? 'mqtts' : 'mqtt';
    const broker = `${protocol}://${mqttConfig.host}:${mqttConfig.port}`;
    
    // 基础连接选项
    const connectOptions = {
      clientId: mqttConfig.clientId,
      username: mqttConfig.username,
      password: mqttConfig.password,
      keepalive: mqttConfig.keepalive,
      reconnectPeriod: mqttConfig.reconnectPeriod,
      connectTimeout: mqttConfig.connectTimeout,
      clean: mqttConfig.clean,
      protocolVersion: mqttConfig.protocolVersion,
      // 添加遗嘱消息（Will Message）
      will: {
        topic: `clients/${mqttConfig.clientId}/status`,
        payload: JSON.stringify({
          clientId: mqttConfig.clientId,
          status: 'offline',
          timestamp: new Date().toISOString(),
          reason: 'unexpected_disconnect'
        }),
        qos: 1,
        retain: true
      }
    };

    // 如果启用SSL/TLS，添加证书配置
    if (mqttConfig.secure) {
      connectOptions.rejectUnauthorized = mqttConfig.rejectUnauthorized;
      
      // 添加证书配置（如果提供）
      if (mqttConfig.ca) {
        connectOptions.ca = mqttConfig.ca;
      }
      if (mqttConfig.cert) {
        connectOptions.cert = mqttConfig.cert;
      }
      if (mqttConfig.key) {
        connectOptions.key = mqttConfig.key;
      }
      
      console.log('使用SSL/TLS连接，证书验证:', mqttConfig.rejectUnauthorized ? '严格模式' : '宽松模式');
    }

    mqttClient = mqtt.connect(broker, connectOptions);

    mqttClient.on('connect', () => {
      console.log('MQTT连接成功');
      addLog('connection', 'connected', null, '连接成功');
      broadcastToClients({ type: 'connection', status: 'connected', message: '连接成功' });

      // 发布在线状态（覆盖遗嘱消息）
      mqttClient.publish(
        `clients/${mqttConfig.clientId}/status`,
        JSON.stringify({
          clientId: mqttConfig.clientId,
          status: 'online',
          timestamp: new Date().toISOString()
        }),
        { qos: 1, retain: true }
      );

      // 重新订阅之前的主题
      subscribedTopics.forEach(topic => {
        mqttClient.subscribe(topic, { qos: 2 });  // 使用QoS 2重新订阅
      });
    });

    mqttClient.on('error', (err) => {
      console.error('MQTT连接错误:', err);
      addLog('connection', 'error', null, '连接错误: ' + err.message);
      broadcastToClients({ type: 'connection', status: 'error', message: '连接错误' });
      backoffUntilTs = Date.now() + 30000;
    });

    mqttClient.on('close', () => {
      console.log('MQTT连接关闭');
      addLog('connection', 'disconnected', null, '连接关闭');
      broadcastToClients({ type: 'connection', status: 'disconnected', message: '连接关闭' });
    });

    mqttClient.on('message', (topic, message, packet) => {
      const messageStr = message.toString();
      const isRetained = packet.retain;
      console.log(`收到消息: ${topic} - ${messageStr} (retained: ${isRetained})`);
      addLog('message', topic, messageStr, `收到消息${isRetained ? '(保留)' : ''}`);
      broadcastToClients({ type: 'message', topic, message: messageStr, source: `收到消息${isRetained ? '(保留)' : ''}` });
    });

  } catch (err) {
    console.error('MQTT连接失败:', err);
    addLog('connection', 'error', null, '连接失败: ' + err.message);
    broadcastToClients({ type: 'connection', status: 'error', message: '连接失败' });
    backoffUntilTs = Date.now() + 30000;
  } finally {
    isConnecting = false;
  }
}

// 断开MQTT连接
function disconnectMQTT() {
  if (mqttClient) {
    mqttClient.end();
    mqttClient = null;
    console.log('MQTT连接已断开');
    addLog('connection', 'disconnected', null, '主动断开连接');
    broadcastToClients({ type: 'connection', status: 'disconnected', message: '主动断开连接' });
  }
}

// 发布消息
function publishMessage(topic, message, qos = 0, retained = false) {
  if (!mqttClient || !mqttClient.connected) {
    console.warn('MQTT客户端未连接，无法发布消息');
    return false;
  }

  try {
    mqttClient.publish(topic, message, { qos, retained });
    console.log(`发布消息成功: ${topic} - ${message} (retained: ${retained})`);
    addLog('publish', topic, message, `发布消息${retained ? '(保留)' : ''}`);
    broadcastToClients({ type: 'publish', topic, message, source: `发布消息${retained ? '(保留)' : ''}` });
    
    // 如果是retained消息，不模拟接收（让MQTT服务器处理）
    // 如果不是retained消息，模拟接收自己发布的消息
    if (!retained && subscribedTopics.has(topic)) {
      console.log(`模拟接收自己发布的消息: ${topic} - ${message}`);
      addLog('message', topic, message, '收到消息');
      broadcastToClients({ type: 'message', topic, message, source: '收到消息' });
    }
    
    return true;
  } catch (err) {
    console.error('发布消息失败:', err);
    addLog('publish', 'error', null, '发布失败: ' + err.message);
    broadcastToClients({ type: 'publish', status: 'error', message: '发布失败' });
    return false;
  }
}

// 订阅主题
function subscribeTopic(topic, qos = 2) {
  if (!mqttClient || !mqttClient.connected) {
    console.warn('MQTT客户端未连接，无法订阅主题');
    return false;
  }

  try {
    // 使用指定的QoS等级订阅
    mqttClient.subscribe(topic, { qos }, (err) => {
      if (err) {
        console.error(`订阅主题失败: ${topic}`, err);
      } else {
        console.log(`订阅主题成功: ${topic} (QoS: ${qos})，等待retained消息...`);

        // 设置超时检查，如果3秒内没有收到retained消息，说明服务器没有保存
        setTimeout(() => {
          console.log(`警告: 订阅 ${topic} 后3秒内未收到retained消息，可能服务器不支持或消息已被清除`);

          // 备选方案：从数据库查询该主题的历史消息
          if (mysqlPool) {
            console.log(`尝试从数据库查询主题 ${topic} 的历史消息...`);
            mysqlPool.query(
              'SELECT * FROM logs WHERE topic = ? AND type = "message" ORDER BY created_at DESC LIMIT 1',
              [topic]
            ).then(([rows]) => {
              if (rows.length > 0) {
                const lastMessage = rows[0];
                console.log(`从数据库找到历史消息: ${topic} - ${lastMessage.message}`);
                addLog('message', topic, lastMessage.message, '收到消息(历史)');
                broadcastToClients({ type: 'message', topic, message: lastMessage.message, source: '收到消息(历史)' });
              } else {
                console.log(`数据库中没有找到主题 ${topic} 的历史消息`);
              }
            }).catch(err => {
              console.warn('从数据库查询历史消息失败:', err.message);
            });
          }
        }, 3000);
      }
    });
    subscribedTopics.add(topic);
    console.log(`订阅主题成功: ${topic} (QoS: ${qos})`);
    addLog('subscribe', topic, null, `订阅主题 (QoS: ${qos})`);
    broadcastToClients({ type: 'subscribe', topic, qos, source: `订阅主题 (QoS: ${qos})` });

    // 注意：retained消息会通过MQTT的message事件自动接收
    // 不需要额外处理，MQTT服务器会自动发送retained消息给新订阅者

    return true;
  } catch (err) {
    console.error('订阅主题失败:', err);
    addLog('subscribe', 'error', null, '订阅失败: ' + err.message);
    broadcastToClients({ type: 'subscribe', status: 'error', message: '订阅失败' });
    return false;
  }
}

// 取消订阅主题
function unsubscribeTopic(topic) {
  if (!mqttClient || !mqttClient.connected) {
    console.warn('MQTT客户端未连接，无法取消订阅');
    return false;
  }

  try {
    mqttClient.unsubscribe(topic);
    subscribedTopics.delete(topic);
    console.log(`取消订阅成功: ${topic}`);
    addLog('unsubscribe', topic, null, '取消订阅');
    broadcastToClients({ type: 'unsubscribe', topic, source: '取消订阅' });
    return true;
  } catch (err) {
    console.error('取消订阅失败:', err);
    addLog('unsubscribe', 'error', null, '取消订阅失败: ' + err.message);
    broadcastToClients({ type: 'unsubscribe', status: 'error', message: '取消订阅失败' });
    return false;
  }
}

// 根路径路由
app.get('/', (req, res) => {
  res.sendFile(path.join(__dirname, 'client-vue/dist/index.html'));
});

// API路由
app.get('/api/status', (req, res) => {
  res.json({
    connected: mqttClient ? mqttClient.connected : false,
    subscribedTopics: Array.from(subscribedTopics),
    config: mqttConfig
  });
});

app.post('/api/connect', async (req, res) => {
  try {
    const { 
      host, 
      port, 
      username, 
      password, 
      clientId, 
      secure, 
      rejectUnauthorized, 
      ca, 
      cert, 
      key 
    } = req.body;
    
    mqttConfig = {
      ...mqttConfig,
      host: host || mqttConfig.host,
      port: port || mqttConfig.port,
      username: username || mqttConfig.username,
      password: password || mqttConfig.password,
      clientId: clientId || mqttConfig.clientId,
      secure: secure !== undefined ? secure : mqttConfig.secure,
      rejectUnauthorized: rejectUnauthorized !== undefined ? rejectUnauthorized : mqttConfig.rejectUnauthorized,
      ca: ca || mqttConfig.ca,
      cert: cert || mqttConfig.cert,
      key: key || mqttConfig.key
    };

    console.log('连接配置:', {
      host: mqttConfig.host,
      port: mqttConfig.port,
      secure: mqttConfig.secure,
      rejectUnauthorized: mqttConfig.rejectUnauthorized,
      hasCertificates: {
        ca: !!mqttConfig.ca,
        cert: !!mqttConfig.cert,
        key: !!mqttConfig.key
      }
    });

    await connectMQTT();
    res.json({ success: true, message: '连接请求已发送' });
  } catch (err) {
    res.json({ success: false, message: '连接失败: ' + err.message });
  }
});

app.post('/api/disconnect', (req, res) => {
  try {
    disconnectMQTT();
    res.json({ success: true, message: '断开连接成功' });
  } catch (err) {
    res.json({ success: false, message: '断开连接失败: ' + err.message });
  }
});

app.post('/api/publish', (req, res) => {
  try {
    const { topic, message, qos = 0, retained = false } = req.body;
    
    if (!topic || !message) {
      return res.json({ success: false, message: '主题和消息不能为空' });
    }

    // 确保消息是字符串类型
    const messageStr = typeof message === 'object' ? JSON.stringify(message) : String(message);
    const success = publishMessage(topic, messageStr, qos, retained);
    res.json({ 
      success, 
      message: success ? '发布成功' : '发布失败' 
    });
  } catch (err) {
    res.json({ success: false, message: '发布失败: ' + err.message });
  }
});

app.post('/api/subscribe', (req, res) => {
  try {
    const { topic, qos = 2 } = req.body;  // 默认QoS 2，确保消息可靠性

    if (!topic) {
      return res.json({ success: false, message: '主题不能为空' });
    }

    const success = subscribeTopic(topic, qos);
    res.json({
      success,
      message: success ? `订阅成功 (QoS: ${qos})` : '订阅失败'
    });
  } catch (err) {
    res.json({ success: false, message: '订阅失败: ' + err.message });
  }
});

app.post('/api/unsubscribe', (req, res) => {
  try {
    const { topic } = req.body;
    
    if (!topic) {
      return res.json({ success: false, message: '主题不能为空' });
    }

    const success = unsubscribeTopic(topic);
    res.json({ 
      success, 
      message: success ? '取消订阅成功' : '取消订阅失败' 
    });
  } catch (err) {
    res.json({ success: false, message: '取消订阅失败: ' + err.message });
  }
});

app.get('/api/logs', async (req, res) => {
  try {
    const limit = parseInt(req.query.limit) || 100;
    let logsToReturn = [];

    // 优先从数据库加载日志（如果数据库可用）
    if (mysqlPool) {
      try {
        const [rows] = await mysqlPool.query(
          'SELECT * FROM logs ORDER BY created_at DESC LIMIT ?',
          [limit]
        );
        logsToReturn = rows.map(row => ({
          timestamp: row.timestamp,
          type: row.type,
          topic: row.topic,
          message: row.message,
          qos: row.qos,
          retained: row.retained,
          source: row.source,
          client_id: row.client_id
        }));
        console.log(`从数据库加载了 ${logsToReturn.length} 条日志`);
      } catch (dbErr) {
        console.warn('从数据库加载日志失败，使用内存日志:', dbErr.message);
        // 如果数据库失败，回退到内存日志
        logsToReturn = logs.slice(0, limit);
      }
    } else {
      // 如果数据库不可用，使用内存日志
      logsToReturn = logs.slice(0, limit);
    }

    res.json({ success: true, logs: logsToReturn });
  } catch (err) {
    res.json({ success: false, message: '获取日志失败: ' + err.message });
  }
});

app.get('/api/logs/type/:type', (req, res) => {
  const type = req.params.type;
  const limit = parseInt(req.query.limit) || 100;
  const logsToReturn = logs.filter(log => log.type === type).slice(0, limit);
  res.json(logsToReturn);
});

app.delete('/api/logs', async (req, res) => {
  try {
    // 清空内存中的日志
    logs.length = 0;
    
    // 同时清空数据库中的日志
    if (mysqlPool) {
      try {
        await mysqlPool.query('DELETE FROM logs');
        console.log('数据库日志已清空');
      } catch (dbErr) {
        console.warn('清空数据库日志失败:', dbErr.message);
        // 继续处理，因为内存日志已经清空
      }
    }
    
    // 广播日志清空事件给所有WebSocket客户端
    broadcastToClients({ type: 'logsCleared', message: '日志已清空' });
    
    res.json({ success: true, message: '日志和数据库信息已清空' });
  } catch (err) {
    res.json({ success: false, message: '清空日志失败: ' + err.message });
  }
});

app.get('/api/stats', (req, res) => {
  const stats = {
    totalLogs: logs.length,
    connectionLogs: logs.filter(log => log.type === 'connection').length,
    publishLogs: logs.filter(log => log.type === 'publish').length,
    subscribeLogs: logs.filter(log => log.type === 'subscribe').length,
    messageLogs: logs.filter(log => log.type === 'message').length
  };
  res.json(stats);
});

// 启动服务器
async function startServer() {
  try {
    // 尝试初始化数据库，但即使失败也继续启动服务器
    await initMySQL();
    
    // 无论数据库初始化是否成功，都启动HTTP服务器
    app.listen(PORT, '0.0.0.0', () => {
      console.log(`服务器运行在端口 ${PORT}`);
      console.log(`WebSocket服务器运行在端口 ${WS_PORT}`);
      console.log(`本地访问: http://localhost:${PORT}`);
      console.log(`公网访问: http://8.149.243.80:${PORT}`);
      console.log(`数据库状态: ${mysqlPool ? '已连接' : '未连接，使用内存日志'}`);
    });
  } catch (err) {
    console.error('服务器启动时发生错误:', err);
    // 即使有错误，也尝试启动服务器
    app.listen(PORT, '0.0.0.0', () => {
      console.log(`服务器运行在端口 ${PORT} (部分功能可能受限)`);
      console.log(`WebSocket服务器运行在端口 ${WS_PORT}`);
      console.log(`本地访问: http://localhost:${PORT}`);
      console.log(`公网访问: http://8.149.243.80:${PORT}`);
    });
  }
}

startServer();
